package com.yet.container.remote.netty.server;

import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.core.Ordered;

import com.yet.container.registry.LocalServiceStateHolder;
import com.yet.container.registry.RegistryMeta.ServiceMeta;
import com.yet.dto.Request;
import com.yet.dto.Response;
import com.yet.exception.RegistryNotFoundException;
import com.yet.exception.RemotingException;
import com.yet.listener.ServerResponseEvent;
import com.yet.listener.ServerStartedEvent;
import com.yet.listener.ServerStopedEvent;
import com.yet.serialization.Serialization;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.DefaultChannelPromise;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.MessageToByteEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;

public class NettyServer implements Server {

	private final static Logger logger = LoggerFactory.getLogger(NettyServer.class);
	private final InetSocketAddress localAddress;
	private final Set<ServiceMeta> serviceMetas = new CopyOnWriteArraySet<>();
	private final Serialization serialization;
	private EventLoopGroup boss;
	private EventLoopGroup worker;
	private Channel channel;
	private ApplicationEventPublisher publisher;

	public NettyServer(InetSocketAddress localAddress, Serialization serialization) {
		this.localAddress = localAddress;
		this.serialization = serialization;
	}

	@Override
	public void export(ServiceMeta serviceMeta) {
		serviceMetas.add(serviceMeta);
	}

	class NettyDecoder extends ByteToMessageDecoder {

		private Class<?> clzz;

		public NettyDecoder(Class<?> clzz) {
			this.clzz = clzz;
		}

		@Override
		protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {

			if (in.readableBytes() < 4) {
				return;
			}
			in.markReaderIndex();
			int dataLength = in.readInt();
			if (dataLength < 0) {
				ctx.close();
			}
			if (in.readableBytes() < dataLength) {
				in.resetReaderIndex();
				return; // fix 1024k buffer splice limix
			}
			byte[] data = new byte[dataLength];
			in.readBytes(data);

			Object obj = serialization.deserialize(data, clzz);
			out.add(obj);
		}

	}

	class NettyEncoder extends MessageToByteEncoder<Response> {

		@Override
		protected void encode(ChannelHandlerContext ctx, Response msg, ByteBuf out) throws Exception {
			byte[] data = serialization.serialize(msg);
			out.writeInt(data.length);
			out.writeBytes(data);
		}
	}

	@ChannelHandler.Sharable
	class NettyServerHandler extends ChannelInboundHandlerAdapter {

		private int loss_connect_time = 0;

		@Override
		public void channelActive(ChannelHandlerContext ctx) throws Exception {
			loss_connect_time = 0;
			super.channelActive(ctx);
		}

		@Override
		public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
			Response response = new Response();
			if (msg instanceof Request) {
				Request req = (Request) msg;
				response.setId(req.getId());
				Object service = LocalServiceStateHolder.getService(getServiceMeta(req));
				if (service != null) {
					Method method = service.getClass().getMethod(req.getMethodName(), req.getParameterTypes());
					Object result = method.invoke(service, req.getParameters());
					response.setResult(result);
				}
				ctx.writeAndFlush(response, new DefaultChannelPromise(ctx.channel()).addListener(f -> {
					publisher.publishEvent(new ServerResponseEvent(localAddress, req, response));
				}));
			} else {
				logger.debug("No support type:", msg.getClass());
				ctx.writeAndFlush(response);
			}
		}

		@Override
		public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
			if (evt instanceof IdleStateEvent) {
				IdleStateEvent event = (IdleStateEvent) evt;
				if (event.state() == IdleState.READER_IDLE) {
					loss_connect_time++;
					if (loss_connect_time > 2) {
						System.out.println("Close:" + ctx.channel().id().asShortText());
						logger.info("失去链接channel：{}，关闭处理", localAddress);
						ctx.channel().close();
					}
				}
			} else {
				super.userEventTriggered(ctx, evt);
			}
		}
	}

	@Override
	public void start() throws RuntimeException {
		boss = new NioEventLoopGroup(1);
		worker = new NioEventLoopGroup();
		ChannelFuture future = new ServerBootstrap().group(boss, worker).localAddress(localAddress)
				.handler(new LoggingHandler(LogLevel.INFO)).childOption(ChannelOption.TCP_NODELAY, true)
				.childOption(ChannelOption.SO_KEEPALIVE, true).channel(NioServerSocketChannel.class)
				.childHandler(new ChannelInitializer<SocketChannel>() {
					@Override
					protected void initChannel(SocketChannel ch) throws Exception {

						ch.pipeline().addLast(new IdleStateHandler(2, 0, 0, TimeUnit.MINUTES))
								.addLast(new NettyDecoder(Request.class)).addLast(new NettyEncoder())
								.addLast(new NettyServerHandler());
					}
				}).bind();
		try {
			channel = future.sync().addListener(f -> {
				if (f.isSuccess()) {
					logger.info("服务启动成功 : 地址 {} 服务 [{}]", localAddress, serviceMetas);
					publisher.publishEvent(new ServerStartedEvent(localAddress, channel));
				}
			}).channel();
		} catch (InterruptedException e) {
			stop();
			throw new RemotingException("Netty服务启动失败", e);
		}
	}

	@Override
	public void stop() {
		boss.shutdownGracefully();
		worker.shutdownGracefully();
		if (channel != null && channel.isOpen()) {
			channel.close();
		}
		publisher.publishEvent(new ServerStopedEvent(serviceMetas));
	}

	@Override
	public int getOrder() {
		return Ordered.HIGHEST_PRECEDENCE + 300;
	}

	private ServiceMeta getServiceMeta(Request request) {
		return serviceMetas.stream().filter(i -> i.equals(request.getServiceMeta())).findFirst()
				.orElseThrow(() -> new RegistryNotFoundException("找不到服务" + request.getServiceMeta()));
	}

	public void setPublisher(ApplicationEventPublisher publisher) {
		this.publisher = publisher;
	}
}
